package com.sunl.mall.ali.mq.consumer;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.MessageSelector;
import com.aliyun.openservices.ons.api.bean.OrderConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.sunl.mall.ali.mq.api.consumer.RocketMqOrderMessageListener;
import com.sunl.mall.ali.mq.model.RocketMqMessageBody;
import com.sunl.mall.ali.mq.model.order.RocketMqOrderMessageContext;
import com.sunl.mall.ali.mq.topic.SubscribeTopic;
import com.sunl.mall.ali.mq.util.MqContextUtil;
import com.sunl.mall.ali.mq.util.MqUtil;
import com.sunl.mall.common.api.topic.MqTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class RocketMqOrderConsumerBean extends OrderConsumerBean implements SubscriptionRelationChecker, SubscriptionLogger {
    private final static Logger log = LoggerFactory.getLogger(RocketMqConsumerBean.class);

    private String consumerGroupId;

    public RocketMqOrderConsumerBean() {
    }

    @Override
    public void start() {
        this.consumerGroupId = MqUtil.getConsumerId(this.getProperties());
        log.info("########## Start to create orderConsumer. consumerGroupId=[{}]", this.consumerGroupId);
        this.check(this.consumerGroupId);
        super.start();
    }


    @Override
    public void subscribe(final String topic, final String tags, final MessageOrderListener listener) {
        log.info("########## Start to subscribe topic. consumerGroupId=[{}] topic=[{}] tags=[{}]", this.consumerGroupId, topic, tags);
        super.subscribe(topic, tags, listener);
    }

    /**
     * 通過<code>selector</code>订阅，仅购买阿里<code>RocketMQ</code>铂金版可以使用，且不允许公网访问
     *
     * @param topic    topic
     * @param selector selector
     * @param listener listener
     */
    @Override
    public void subscribe(String topic, MessageSelector selector, MessageOrderListener listener) {
        log.info("########## Start to subscribe topic. consumerGroupId=[{}] topic=[{}] selector=[{}]", this.consumerGroupId, topic, selector.getSubExpression());
        super.subscribe(topic, selector, listener);
    }

    /**
     * 设置订阅关系
     * 1.将自定义 listener 转成 sdk中的 {@link MessageListener}
     * 2.调用父类中设置订阅关系逻辑
     *
     * @param subscriptionTable
     * @param <T>
     */
    public <T> void setSubscriptionRelation(final List<RocketMqOrderMessageListener<T>> subscriptions) {
        final Map<Subscription, MessageOrderListener> table = new HashMap<>(subscriptions.size());
        for (RocketMqOrderMessageListener<T> subscription : subscriptions) {
            MqTopic rockMqTopic = subscription.subscribe();
            if (rockMqTopic == null) {
                throw new IllegalArgumentException("MqConsumer[" + subscription.getClass() + "] subscribe topic must not null, and must implement method[subscribe]");
            }
            // 将自定义listener转成sdk中的listener
            MessageOrderListener exist = table.putIfAbsent(new SubscribeTopic(rockMqTopic), ((originMsg, originContext) -> {
                RocketMqMessageBody<T> record = null;
                final RocketMqOrderMessageContext context = MqContextUtil.getOrderMessageContext(originMsg);
                try {
                    record = subscription.deserialize(originMsg.getBody(), context);
                    this.logDeserializeResult(context, consumerGroupId, record);
                    return subscription.call(record, context);
                } catch (Throwable throwable) {
                    this.logThrowable(context, consumerGroupId, record, throwable);
                    return OrderAction.Suspend;
                }
            }));
            if (exist != null) {
                throw new IllegalArgumentException("MqConsumer[" + subscription.getClass() + "] subscribe topic[" + rockMqTopic.getTopic() + "] tag[" + rockMqTopic.getExpression() + "] is exist!");
            }
        }
        // 设置订阅关系
        super.setSubscriptionTable(table);
    }
}
